package POE::Component::Gearman::Connection;

use 5.006;
use strict;
use warnings FATAL => 'all';

#use Smart::Comments;

use IO::Socket;
use POE qw(
    Component::Gearman::Constants
    Wheel::SocketFactory
    Wheel::ReadWrite
    Filter::Gearman
);

=head1 NAME

POE::Component::Gearman::Connection - Asynchronous Gearman connection for POE applications 

=head1 VERSION

Version 0.01

=cut

our $VERSION = '0.01';


=head1 SYNOPSIS

    use POE qw(Component::Gearman::Connection);

    my $connection = new POE::Component::Gearman::Connection {
        JobServers  => [ '127.0.0.1', '192.168.0.1:123' ],
    };

    run POE::Kernel;

=head1 EXPORT

A list of functions that can be exported.  You can delete this section
if you don't export anything, such as for a purely object-oriented module.

=head1 SUBROUTINES/METHODS

=head2 spawn

=cut

sub new {
    shift if defined $_[0] and $_[0] eq __PACKAGE__;
    my ($ip, $port) = @_;

    return new POE::Wheel::SocketFactory(
        SocketType      => SOCK_STREAM,
        SocketDomain    => PF_INET,
        SocketProtocol  => 'tcp',
        RemoteAddress   => $ip,
        RemotePort      => $port || 4730,
        SuccessEvent    => 'on_tcp_connected',
        FailureEvent    => "on_connect_error",
    );
}

=head2 

=cut

sub on_tcp_connected {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my ($socket, $ip, $port, $sock_factory_wheel_id) = @_[ ARG0..$#_ ];

    my $cfg = $heap->{cfg};
    my $box = $heap->{box};

    my $read_write_wheel = new POE::Wheel::ReadWrite(
        Handle      => $socket,
        InputEvent  => 'on_recv_command',
        ErrorEvent  => 'on_read_write_error',
        Filter      => new POE::Filter::Gearman,
    );

    # 在建连阶段，connections 里面存的是 Wheel::SocketFactory 的 ID，
    # 但当建连完成后，基本上用得着的只有 Wheel::ReadWrite 的 ID，
    # （考虑 InputEvent 事件的回调参数）
    # 所以这里需要拆卸下来重新悬挂一下。
    # 好在所有的 Wheel 的 ID 都在同一个名字空间下，所以不会重复。
    my $connDesc = delete $heap->{connections}{$sock_factory_wheel_id};
    $connDesc->{state}      = 'connected';
    $connDesc->{rw_wheel}   = $read_write_wheel;
    $connDesc->{ip}         = inet_ntoa($ip);
    my $id                  = $read_write_wheel->ID;
    $heap->{connections}{$id} = $connDesc;

    ### connected: $connDesc->{connStr}

    $kernel->yield( 'on_connection_ready' => $id );
}

=head2 

=cut

sub on_connect_error {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my ($op, $errNum, $errStr, $wheelID) = @_[ARG0..ARG3];

    my $cfg = $heap->{cfg};
    my $box = $heap->{box};

    warn "Wheel $wheelID generated $op error $errNum: $errStr\n";

    my $connDesc = $heap->{connections}{$wheelID};
    delete $connDesc->{sock_wheel};
    $connDesc->{state} = 'wait4reconnect';

    my $retryInterval = $cfg->{ReconnectInterval} || 60;
    $kernel->delay( reconnect => $retryInterval => $wheelID );
}

=head2 

=cut

sub reconnect {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $sock_factory_wheel_id = $_[ ARG0 ];

    my $connDesc = delete $heap->{connections}{$sock_factory_wheel_id};
    my $host = $connDesc->{host};
    my $port = $connDesc->{port};

    my $wheel = new POE::Component::Gearman::Connection( $host, $port );
    my $id = $wheel->ID;

    $connDesc->{sock_wheel} = $wheel;
    $connDesc->{state} = 'reconnect';

    $heap->{connections}{$id} = $connDesc;
}

=head2 

=cut

sub on_read_write_error {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my $cfg = $heap->{cfg};
    my $box = $heap->{box};
    my ($op, $errNum, $errStr, $wheelID) = @_[ARG0..ARG3];

    warn "Wheel $wheelID generated $op error $errNum: $errStr\n";

    my $connDesc = $heap->{connections}{$wheelID};
    delete $connDesc->{sock_wheel};
    delete $connDesc->{rw_wheel};
    delete $connDesc->{ip};
    $connDesc->{state} = 'wait4reconnect';

    my $retryInterval = $cfg->{ReconnectInterval} || 60;
    $kernel->delay( reconnect => $retryInterval => $wheelID );
}

=head2 

=cut

sub on_recv_command {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my ($input, $wheelID) = @_[ ARG0..$#_ ];
    my $cfg = $heap->{cfg};
    my $box = $heap->{box};

    ### recv: $input

    if ( not defined $input->{code} ){
        warn 'Unkown command.';
        return;
    }

    if ( $input->{way} eq 'RES' ){
        $kernel->yield( 'on_' . $input->{code}, $input->{data}, $wheelID );
    }

    return;
}

=head2 

=cut

sub send_command {
    my ($kernel, $heap) = @_[ KERNEL, HEAP ];
    my ($wheelID, $code, @data) = @_[ ARG0..$#_ ];

    my $cfg = $heap->{cfg};
    my $box = $heap->{box};

    my $pkt = {
        way     => 'REQ',
        code    => $code,
        data    => [ @data ],
    };

    ### send: $pkt

    my $wheel = $heap->{connections}{$wheelID}->{rw_wheel};
    $wheel->put( $pkt );
}

1;
